package mr.stocks.question1;

import java.io.IOException;
import java.util.Arrays;
import java.util.Calendar;

import mr.stocks.avro.model.StocksKeyAvro;
import mr.stocks.avro.model.StocksReportAvro;
import mr.stocks.avro.model.StocksValuesAvro;
import mr.stocks.model.StocksInputWritable;

import org.apache.avro.mapred.AvroKey;
import org.apache.avro.mapred.AvroValue;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.mapreduce.Mapper;

public class StocksAvroMapper extends
		Mapper<LongWritable, StocksInputWritable,AvroKey<StocksKeyAvro>, AvroValue<StocksReportAvro>> {
	private StocksKeyAvro stocksKey = new StocksKeyAvro();
	private StocksReportAvro stocksReport = new StocksReportAvro();
	private StocksValuesAvro stocksValue = new StocksValuesAvro();
	
	private AvroKey<StocksKeyAvro> key = new AvroKey<StocksKeyAvro>();
	private AvroValue<StocksReportAvro> value = new AvroValue<StocksReportAvro>();

	private Calendar cal = Calendar.getInstance();

	@Override
	protected void setup(Context context) throws IOException,
			InterruptedException {
	}

	@Override
	protected void map(LongWritable lineNum, StocksInputWritable inputStock,
			Context context) throws IOException, InterruptedException {

		cal.setTimeInMillis(inputStock.getTimeStamp());
		
		stocksKey.setYear(Integer.toString(cal.get(Calendar.YEAR)));
		stocksKey.setMonth(Integer.toString(cal.get(Calendar.MONTH)+1));
		stocksKey.setDay(Integer.toString(cal.get(Calendar.DAY_OF_MONTH)));
		key.datum(stocksKey);

		stocksReport.setYearMonthDay(stocksKey);
		stocksValue.setStock(inputStock.getStockSymbol());
		stocksValue.setTimestamp(inputStock.getTimeStamp());
		stocksValue.setPrice(inputStock.getPrice());
		stocksReport.setStocks(Arrays.asList(stocksValue));
		value.datum(stocksReport);
		
		context.write(key, value);
	}
}
